package com.xuegao.xuegaospringboottest.controller;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.ObjectUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
public class KafkaController {
    private static final Logger log = LoggerFactory.getLogger(KafkaController.class);

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    private static final String KAFKA_TOPIC = "test-send";

    @RequestMapping("/test/send")
    public String send() {
        ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(KAFKA_TOPIC, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa");
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("onFailure");
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("onSuccess");
            }
        });
        return "success";
    }

    @KafkaListener(topics = {KAFKA_TOPIC}, containerFactory = "listenerContainerFactory")
    public void kafkaListener(List<ConsumerRecord<?, String>> record, Acknowledgment acknowledgment) {
        ObjectMapper objectMapper = new ObjectMapper();
        log.info("[xue-gao-spring-boot-test][KafkaController][kafkaListener][begin]");
        if (ObjectUtils.isEmpty(record)) {
            log.info("[xue-gao-spring-boot-test][KafkaController][kafkaListener][is null]");
            acknowledgment.acknowledge();
        }
        try {
            for (ConsumerRecord<?, String> consumerRecord : record) {
                Object key = consumerRecord.key();
                log.info("[xue-gao-spring-boot-test][KafkaController][kafkaListener][key={}]", objectMapper.writeValueAsString(key));
                String value = consumerRecord.value();
                log.info("[xue-gao-spring-boot-test][KafkaController][kafkaListener][value={}]", value);
            }
        } catch (Exception e) {
            log.info("[xue-gao-spring-boot-test][KafkaController][kafkaListener][e]", e);
        }
        acknowledgment.acknowledge();
        log.info("[xue-gao-spring-boot-test][KafkaController][kafkaListener][end]");
    }
}
